package com.bujunjie.study.realtime.common.base;

import com.bujunjie.study.realtime.common.util.FlinkSourceUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * <p>flink 作业抽象出来的作业类</p>
 *
 * @author bu.junjie
 * @version 1.0.0
 * @createTime 2025/9/12 14:28
 */
public abstract class BaseApp {

    /**
     * <p>启动 flink 作业</p>
     *
     * @param port         端口
     * @param parallelism  作业的并行度
     * @param ckAndGroupId kafka 的 groupId
     * @param topic        kafka 的 topic
     * @author bu.junjie
     * @date 2025/9/12 14:29
     */
    public void start(int port, int parallelism, String ckAndGroupId, String topic) throws Exception {
        // 1.1 指定流处理环境
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", port);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        // 1.2 设置并行度
        env.setParallelism(parallelism);

        // 2. 检查点的相关设置
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
       /* // 2.1 开启检查点
        // 2.2 设置检查点超时时间,为1分钟 当前都是毫秒级别的时间戳
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        // 2.3 设置 job 取消后检查点是否保留
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 2.4 设置检查点之间最小时间间隔，当前都是毫秒级别的时间戳
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        // 2.5 设置重启策略：时间间隔为30天，即在30天内如果失败次数超过3次，则作业将不再重启
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30), Time.seconds(3)));

        // 2.6 设置状态后端以及检查点存储路径
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop-master:8020/flink/ck" + ckAndGroupId);
        // 2.7 设置操作 hadoop 的用户名
        System.setProperty("HADOOP_USER_NAME", "bujunjie");*/

        // 3. 从 kafka 中的主题中读取业务数据
        // 3.1 声明消费者的主题以及消费者组
        // 3.2 创建消费者对象
        KafkaSource<String> kafkaSource = FlinkSourceUtil.getKafkaSource(topic, ckAndGroupId);
        //3.3 消费数据 封装为流
        DataStreamSource<String> kafkaStrDs = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka-source");
        // 4. 处理 kafka 的流数据
        this.handle(env, kafkaStrDs);
        // 5.提交作业
        env.execute();
    }

    /**
     * <p>处理流数据</p>
     *
     * @param env        执行环境
     * @param kafkaStrDS kafka 流
     * @author bu.junjie
     * @date 2025/9/12 15:22
     */
    public abstract void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS);
}
